package drds.server.net.packet;

import java.nio.ByteBuffer;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import drds.server.config.Fields;
import drds.server.memory.unsafe.row.UnsafeRow;
import drds.server.net.SessionContext$LifeCycle;
import drds.server.util.ByteUtil;
import drds.server.util.DateUtil;

/**
 * ProtocolBinary::ResultsetRow: row of a binary resultset (COM_STMT_EXECUTE)
 * 
 * Payload 1 packet header [00] string[$len] NULL-bitmap, length: (column_count
 * + 7 + 2) / 8 string[$len] values
 * 
 * A Binary Protocol Resultset Row is made up of the NULL bitmap containing as
 * many bits as we have columns in the resultset + 2 and the values for columns
 * that are not NULL in the Binary Protocol Value format.
 * 
 * @see
 * @http://dev.mysql.com/doc/internals/en/binary-protocol-resultset-row.html 
 *                                                                           #packet
 *                                                                           -
 *                                                                           ProtocolBinary
 *                                                                           ::
 *                                                                           ResultsetRow
 * @see @http://dev.mysql.com/doc/internals/en/binary-protocol-value.html
 * @author 曹正辉<330937205@qq.com>
 * 
 */
public class BinaryRowDataPacket extends Packet {
	private static final Logger LOGGER = LoggerFactory.getLogger(BinaryRowDataPacket.class);
	public int fieldCount;
	public List<byte[]> fieldValues;
	public byte packetHeader = (byte) 0;

	public List<FieldPacket> fieldPacketList;

	public BinaryRowDataPacket() {
	}

	/**
	 * 从UnsafeRow转换成BinaryRowDataPacket
	 * 
	 * 说明: 当开启<b>isOffHeapuseOffHeapForMerge</b>参数时,会使用UnsafeRow封装数据,
	 * 因此需要从这个对象里面将数据封装成BinaryRowDataPacket
	 */
	public void read(List<FieldPacket> fieldPacketList, UnsafeRow unsafeRow) {
		this.fieldPacketList = fieldPacketList;
		this.fieldCount = unsafeRow.numFields();
		this.fieldValues = new ArrayList<byte[]>(fieldCount);

		for (int i = 0; i < this.fieldCount; i++) {
			byte[] bytes = unsafeRow.getBinary(i);
			FieldPacket fieldPacket = fieldPacketList.get(i);
			convert(bytes, fieldPacket);
		}
	}

	/**
	 * 从RowDataPacket转换成BinaryRowDataPacket
	 * 
	 * @param fieldPacketList
	 *            字段包集合
	 * @param rowDataPacket
	 *            文本协议行数据包
	 */
	public void read(List<FieldPacket> fieldPacketList, RowDataPacket rowDataPacket) {
		this.fieldPacketList = fieldPacketList;
		this.fieldCount = rowDataPacket.fieldCount;
		this.fieldValues = new ArrayList<byte[]>(fieldCount);

		List<byte[]> fieldValueList = rowDataPacket.fieldBytesList;
		for (int i = 0; i < fieldCount; i++) {
			byte[] bytes = fieldValueList.get(i);
			FieldPacket fieldPacket = fieldPacketList.get(i);
			if (bytes == null) { // 字段值为null,根据协议规定存储nullBitMap
				this.fieldValues.add(bytes);
			} else {
				convert(bytes, fieldPacket);
			}
		}
	}

	/**
	 * 从RowDataPacket的fieldValue的数据转化成BinaryRowDataPacket的fieldValue数据
	 */
	private void convert(byte[] fv, FieldPacket fieldPacket) {

		int fieldType = fieldPacket.type;
		switch (fieldType) {
		case Fields.FIELD_TYPE_STRING:
		case Fields.FIELD_TYPE_VARCHAR:
		case Fields.FIELD_TYPE_VAR_STRING:
		case Fields.FIELD_TYPE_BIT:
		case Fields.FIELD_TYPE_DECIMAL:
		case Fields.FIELD_TYPE_NEW_DECIMAL:
			// Fields
			// value (lenenc_str) -- string

			// Example
			// 03 66 6f 6f -- string = "foo"
			this.fieldValues.add(fv);
			break;
		case Fields.FIELD_TYPE_LONG:
			// Fields
			// value (8) -- integer

			// Example
			// 01 00 00 00 00 00 00 00 -- int64 = 1
			long longVar = ByteUtil.getLong(fv);
			this.fieldValues.add(ByteUtil.getBytes(longVar));
			break;
		case Fields.FIELD_TYPE_INT:
		case Fields.FIELD_TYPE_INT24:
			// Fields
			// value (4) -- integer

			// Example
			// 01 00 00 00 -- int32 = 1
			int intVar = ByteUtil.getInt(fv);
			this.fieldValues.add(ByteUtil.getBytes(intVar));
			break;
		case Fields.FIELD_TYPE_SHORT:
		case Fields.FIELD_TYPE_YEAR:
			// Fields
			// value (2) -- integer

			// Example
			// 01 00 -- int16 = 1
			short shortVar = ByteUtil.getShort(fv);
			this.fieldValues.add(ByteUtil.getBytes(shortVar));
			break;
		case Fields.FIELD_TYPE_TINY:
			// Fields
			// value (1) -- integer

			// Example
			// 01 -- int8 = 1
			int tinyVar = ByteUtil.getInt(fv);
			byte[] bytes = new byte[1];
			bytes[0] = (byte) tinyVar;
			this.fieldValues.add(bytes);
			break;
		case Fields.FIELD_TYPE_DATE:
			try {
				Date dateVar = DateUtil.parseDate(ByteUtil.getDate(fv), DateUtil.YYYY_MM_dd);
				this.fieldValues.add(ByteUtil.getBytes(dateVar, false));
			} catch (ParseException e) {
				LOGGER.error("error", e);
			}
			break;
		case Fields.FIELD_TYPE_DATETIME:
		case Fields.FIELD_TYPE_TIMESTAMP:
			String dateStr = ByteUtil.getDate(fv);
			Date dateTimeVar = null;
			try {
				if (dateStr.indexOf(".") > 0) {
					dateTimeVar = DateUtil.parseDate(dateStr, DateUtil.YYYY_MM_dd_HH_mm_ss_SSSSSS);
					this.fieldValues.add(ByteUtil.getBytes(dateTimeVar, false));
				} else {
					dateTimeVar = DateUtil.parseDate(dateStr, DateUtil.YYYY_MM_dd_HH_mm_ss);
					this.fieldValues.add(ByteUtil.getBytes(dateTimeVar, false));
				}
			} catch (ParseException e) {
				LOGGER.error("error", e);
			}
			break;
		case Fields.FIELD_TYPE_TIME:
			String timeStr = ByteUtil.getTime(fv);
			Date timeVar = null;
			try {
				if (timeStr.indexOf(".") > 0) {
					timeVar = DateUtil.parseDate(timeStr, DateUtil.HHH_mm_ss_SSSSSS);
					this.fieldValues.add(ByteUtil.getBytes(timeVar, true));
				} else {
					timeVar = DateUtil.parseDate(timeStr, DateUtil.HHH_mm_ss);
					this.fieldValues.add(ByteUtil.getBytes(timeVar, true));
				}
			} catch (ParseException e) {
				LOGGER.error("error", e);
			}
			break;
		}

	}

	public void write(SessionContext$LifeCycle conn) {

		int size = computePacketSize();
		int packetHeaderSize = conn.getPacketHeaderSize();
		int totalSize = size + packetHeaderSize;
		ByteBuffer bb = null;

		bb = conn.getProcessor().getByteBufferPool().allocate(totalSize);

		ByteBufferUtil.writeInt3(bb, computePacketSize());
		bb.put(packetId);
		bb.put(packetHeader); // packet header [00]
		for (int i = 0; i < fieldCount; i++) { // values
			byte[] fv = fieldValues.get(i);
			if (fv != null) {
				FieldPacket fieldPk = this.fieldPacketList.get(i);
				int fieldType = fieldPk.type;
				switch (fieldType) {
				case Fields.FIELD_TYPE_STRING:
				case Fields.FIELD_TYPE_VARCHAR:
				case Fields.FIELD_TYPE_VAR_STRING:
				case Fields.FIELD_TYPE_BIT:
				case Fields.FIELD_TYPE_DECIMAL:
				case Fields.FIELD_TYPE_NEW_DECIMAL:
					// 长度编码的字符串需要一个字节来存储长度(0表示空字符串)
					ByteBufferUtil.writeLength(bb, fv.length);
					break;
				default:
					break;
				}
				if (fv.length > 0) {
					bb.put(fv);
				}
			}
		}
		conn.write(bb);

	}

	@Override
	public ByteBuffer write(ByteBuffer byteBuffer, SessionContext$LifeCycle sessionContext, boolean writeSocketIfFull) {
		int size = computePacketSize();
		int packetHeaderSize = sessionContext.getPacketHeaderSize();
		int totalSize = size + packetHeaderSize;
		byteBuffer = sessionContext.checkWriteByteBuffer(byteBuffer, totalSize, writeSocketIfFull);
		ByteBufferUtil.writeInt3(byteBuffer, size);
		byteBuffer.put(packetId);
		byteBuffer.put(packetHeader); // packet header [00]
		for (int i = 0; i < fieldCount; i++) { // values
			byte[] fv = fieldValues.get(i);
			if (fv != null) {
				FieldPacket fieldPk = this.fieldPacketList.get(i);
				int fieldType = fieldPk.type;
				switch (fieldType) {
				case Fields.FIELD_TYPE_STRING:
				case Fields.FIELD_TYPE_VARCHAR:
				case Fields.FIELD_TYPE_VAR_STRING:
				case Fields.FIELD_TYPE_BIT:
				case Fields.FIELD_TYPE_DECIMAL:
				case Fields.FIELD_TYPE_NEW_DECIMAL:
					// 长度编码的字符串需要一个字节来存储长度(0表示空字符串)
					ByteBufferUtil.writeLength(byteBuffer, fv.length);
					break;
				default:
					break;
				}
				if (fv.length > 0) {
					byteBuffer.put(fv);
				}
			}
		}
		return byteBuffer;
	}

	@Override
	public int computePacketSize() {
		int size = 0;
		size = size + 1;
		for (int i = 0, n = fieldValues.size(); i < n; i++) {
			byte[] value = fieldValues.get(i);
			if (value != null) {
				FieldPacket fieldPk = this.fieldPacketList.get(i);
				int fieldType = fieldPk.type;
				switch (fieldType) {
				case Fields.FIELD_TYPE_STRING:
				case Fields.FIELD_TYPE_VARCHAR:
				case Fields.FIELD_TYPE_VAR_STRING:
				case Fields.FIELD_TYPE_BIT:
				case Fields.FIELD_TYPE_DECIMAL:
				case Fields.FIELD_TYPE_NEW_DECIMAL:
					/*
					 * 长度编码的字符串需要计算存储长度, 根据mysql协议文档描述 To convert a
					 * length-encoded integer into its numeric value, check the
					 * first byte: If it is < 0xfb, treat it as a 1-byte
					 * integer. If it is 0xfc, it is followed by a 2-byte
					 * integer. If it is 0xfd, it is followed by a 3-byte
					 * integer. If it is 0xfe, it is followed by a 8-byte
					 * integer.
					 */
					if (value.length != 0) {
						/*
						 * 长度编码的字符串需要计算存储长度,不能简单默认只有1个字节是表示长度,当数据足够长,占用的就不止1个字节
						 */
						// size = size + 1 + value.length;
						size = size + ByteBufferUtil.getLength(value);
					} else {
						size = size + 1; // 处理空字符串,只计算长度1个字节
					}
					break;
				default:
					size = size + value.length;
					break;
				}
			}
		}
		return size;
	}

}
